package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.entity.SnackMQRaftLogEntry;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.protocol.OperationType;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class DeleteTopicHandler extends AbstractLeaderRequiredConsoleHandler {

    private final SnackMQRaftNodeManager dataRaftNodeManager;

    public DeleteTopicHandler(MetadataManager metadataManager, SnackMQRaftNodeManager dataRaftNodeManager, AuthorizationService authorizationService) {
        super(metadataManager, authorizationService);
        this.dataRaftNodeManager = dataRaftNodeManager;
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.DELETE_TOPIC_REQUEST;
    }

    @Override
    protected void doHandle(RequestContext ctx, Message request) {
        try {
            String topicName = request.getProperty("topicName");
            if (topicName == null || topicName.isEmpty()) {
                throw new IllegalArgumentException("topicName property is required.");
            }
            // 【核心修复】颠倒操作顺序，先清理数据，再删除元数据，以避免竞态条件。
            // 步骤 1: 创建一个特殊的消息作为数据清理指令，并提议清理数据状态
            Message cleanupMessage = Message.builder().topic(topicName).build();
            SnackMQRaftLogEntry cleanupLog = new SnackMQRaftLogEntry(OperationType.CLEAN_TOPIC_DATA, cleanupMessage);
            dataRaftNodeManager.apply(cleanupLog, dataStatus -> {
                if (dataStatus.isOk()) {
                    log.info("Successfully proposed data cleanup for topic '{}'. Now proposing metadata deletion.", topicName);
                    // 步骤 2: 数据清理提议成功后，提议删除元数据
                    metadataManager.deleteTopic(topicName, metaStatus -> {
                        if (metaStatus.isOk()) {
                            log.info("Successfully proposed metadata deletion for topic '{}'. Operation complete.", topicName);
                            Message response = Message.builder()
                                    .type(MessageType.RESPONSE)
                                    .messageId(request.getMessageId())
                                    .status(MessageStatus.SUCCESS)
                                    .build();
                            ctx.writeResponse(response);
                        } else {
                            log.error("CRITICAL: Data for topic '{}' was cleaned, but failed to delete its metadata: {}", topicName, metaStatus.getErrorMsg());
                            Message response = Message.builder()
                                    .type(MessageType.RESPONSE)
                                    .messageId(request.getMessageId())
                                    .status(MessageStatus.FAILED)
                                    .errorMessage("Data cleaned, but failed to delete metadata: " + metaStatus.getErrorMsg())
                                    .build();
                            ctx.writeResponse(response);
                        }
                    });
                } else {
                    log.error("Failed to propose data cleanup for topic '{}': {}", topicName, dataStatus.getErrorMsg());
                    Message response = Message.builder()
                            .type(MessageType.RESPONSE)
                            .messageId(request.getMessageId())
                            .status(MessageStatus.FAILED)
                            .errorMessage("Failed to clean up topic data: " + dataStatus.getErrorMsg())
                            .build();
                    ctx.writeResponse(response);
                }
            });
        } catch (Exception e) {
            log.error("Error processing delete topic request", e);
            Message response = Message.builder()
                    .type(MessageType.RESPONSE)
                    .messageId(request.getMessageId())
                    .status(MessageStatus.FAILED)
                    .errorMessage("Internal server error: " + e.getMessage())
                    .build();
            ctx.writeResponse(response);
        }
    }
}